package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author yangjiming
 * @create 2021-04-22 19:00
 */
public class Demo03DianXin {
    private Connection conn;
    private Admin admin;

    @Before
    public void init() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void loadDataToHBase() throws IOException, ParseException {
        if (!admin.tableExists(TableName.valueOf("dianxin_v10"))) {
            HTableDescriptor dianxin = new HTableDescriptor(TableName.valueOf("dianxin_v10"));
            dianxin.addFamily(new HColumnDescriptor("info").setMaxVersions(10));
            admin.createTable(dianxin);
        }

        BufferedReader br = new BufferedReader(new FileReader("data/DIANXIN.csv"));
        String line = br.readLine();

        Table dianxin = conn.getTable(TableName.valueOf("dianxin_v10"));
        int cnt = 0;
        int sum = 0;
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
        while (line != null) {
            String[] splits = line.split(",");
            String mdn = splits[0];
            long start_time = sdf.parse(splits[1]).getTime();


            // 经度
            String longitude = splits[4];
            // 纬度
            String latitude = splits[5];
            String pos = longitude + "," + latitude;

            Put put = new Put(mdn.getBytes());

            put.addColumn("info".getBytes(), "pos".getBytes(), start_time, pos.getBytes());
            dianxin.put(put);
            line = br.readLine();
            cnt++;
            sum++;
            if (cnt == 1000) {
                cnt = 0;
                System.out.println(sum);
            }

        }
        System.out.println(sum);

    }

    @Test
    public void timeToDate() throws ParseException {
        String time = "20180503174337";
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
        Date parse = sdf.parse(time);
        System.out.println(parse.getTime());
    }

    @Test
    public void getLastPostion() throws IOException {
        Table dianxin = conn.getTable(TableName.valueOf("dianxin"));
        String mdn ="0B7038E24B101454BD60FA625255E7A067CD005E";

        Get get = new Get(mdn.getBytes());
        get.setMaxVersions(10);
        Result rs = dianxin.get(get);

        String rowkey = Bytes.toString(rs.getRow());
        System.out.println("rowkey为：" + rowkey);

        for (Cell listCell : rs.listCells()) {
            String family = Bytes.toString(CellUtil.cloneFamily(listCell));
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(listCell));
            String position = Bytes.toString(CellUtil.cloneValue(listCell));
            System.out.println("family:" + family + ",qualified:" + qualifier + ",position:" + position);

        }


    }

    @After
    public void close() {
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
